package com.pw.gmall.realtime.sink;

import com.alibaba.fastjson.JSONObject;
import com.pw.gmall.realtime.common.Constant;
import com.pw.gmall.realtime.entities.TableProcess;
import com.pw.gmall.realtime.utils.JdbcUtils;
import com.pw.gmall.realtime.utils.RedisUtils;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import redis.clients.jedis.Jedis;

import java.io.IOException;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Locale;
import java.util.Map;

/**
 * @Author: linux_future
 * @since: 2022/3/18
 **/
public class PhoenixSink extends RichSinkFunction<Tuple2<JSONObject, TableProcess>> {

    private Connection conn;
    private ValueState<Boolean> isCreateState;
    private Jedis jedis;

    @Override
    public void open(Configuration parameters) throws Exception {
        System.out.println("open=============================================");
        //保存状态
        conn = JdbcUtils.getJdbcConnect(Constant.PHOENIX_DRIVER, Constant.PHOENIX_URL, null, null);
        //是否已经创建
        isCreateState = getRuntimeContext().getState(new ValueStateDescriptor<Boolean>("isCreateState", Boolean.class));

        jedis = RedisUtils.getClient();
    }

    @Override
    public void invoke(Tuple2<JSONObject, TableProcess> value, Context context) throws Exception {
        System.out.println("HBASE处理数据：======================================");
        //执行方法
        //创建表
        //String sql=" create table if not exists xx ()";
        createTable(value);
        //插入数据
        insertTable(value);
        //更新缓存
        updateCache(value);

    }

    private void updateCache(Tuple2<JSONObject, TableProcess> value) {
        //1.优雅的更新
        JSONObject dim = value.f0;
        TableProcess tp = value.f1;
        //key： table:id
        String key = tp.getSink_table()+":"+ dim.getLong("id");
        if(jedis.exists(key)){
            JSONObject obj = new JSONObject();
            for (Map.Entry<String, Object> entry : obj.entrySet()) {
                obj.put(entry.getKey().toUpperCase(Locale.ROOT),entry.getValue());
            }
            jedis.setex(key,2*24*60*60,obj.toJSONString());
        }

        //2.粗暴更新
        //jedis.del(key);

    }

    private void insertTable(Tuple2<JSONObject, TableProcess> value) throws SQLException {
        JSONObject data = value.f0;
        TableProcess tp = value.f1;
        StringBuilder sql = new StringBuilder();
        sql.append("upsert into ")
                .append(tp.getSink_table())
                .append("(").append(tp.getSink_columns())
                .append(") values(")
                .append(tp.getSink_columns().replaceAll("([^,]+)", "?")).append(")");
        System.out.println("插入语句。。。。" + sql);
        PreparedStatement ps = conn.prepareStatement(sql.toString());
        String[] columns = tp.getSink_columns().split(",");
        for (int i = 0; i < columns.length; i++) {
            String key = columns[i];
            Object v = data.get(key);
            ps.setString(i + 1, v == null ? null : v.toString());
        }
        ps.execute();
        conn.commit();
        ps.close();
    }

    private void createTable(Tuple2<JSONObject, TableProcess> value) {
        TableProcess tp = value.f1;
        try {
            if (isCreateState.value() == null) {

                StringBuilder sql = new StringBuilder();
                sql.append("create table if not exists ").append(tp.getSink_table())
                        .append("(").append(tp.getSink_columns().replaceAll("([^,]+)", "$1 varchar"))
                        .append(", constraint pk primary key(").append(tp.getSink_pk() == null ? "id" : tp.getSink_pk())
                        .append("))");
                System.out.println("建表语句：" + sql);
                PreparedStatement statement = conn.prepareStatement(sql.toString());
                statement.execute();
                conn.commit();
                statement.close();
                isCreateState.update(true);
            }
        } catch (IOException e) {
            e.printStackTrace();
        } catch (SQLException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void close() throws Exception {
        //关闭链接
        if (conn != null && !conn.isClosed()) {
            conn.close();
        }
    }
}
